package org.nbict.iot.trash.mogo;

import org.apache.commons.lang.Validate;
import org.apache.storm.mongodb.common.QueryFilterCreator;
import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.TupleUtils;
import org.bson.Document;
import org.bson.conversions.Bson;

/**
 * Created by songseven on 18/6/20.
 */
@Deprecated
public class MongoCollectionUpdateBolt extends AbstractMongoBolt {

    private QueryFilterCreator queryCreator;
    private MongoMapper mapper;
    private boolean upsert;

    public MongoCollectionUpdateBolt(String url, String username, String
            pasaword, String adminname, String database, String
            collectionName, QueryFilterCreator queryCreator, MongoMapper
            mapper) {
        super(url, username, pasaword, adminname, database, collectionName);
        Validate.notNull(queryCreator, "QueryFilterCreator can not be null");
        Validate.notNull(mapper, "MongoMapper can not be null");
        this.queryCreator = queryCreator;
        this.mapper = mapper;
    }

    @Override
    public void execute(Tuple tuple) {

        if (!TupleUtils.isTick(tuple)) {
            try {
                Document e = this.mapper.toDocument(tuple);
                Bson filter = this.queryCreator.createFilter(tuple);
                this.mongoClient.update(filter, e, this.upsert);
                this.collector.ack(tuple);
            } catch (Exception var4) {
                this.collector.reportError(var4);
                this.collector.fail(tuple);
            }

        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
